package com.passion.mq.queues.queue1;

import com.passion.mq.config.RocketMqConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * 类描述：用户操作消息队列顺序消费监听器
 * 当队列的消息需要顺序消费时，传入该实现类
 * @author ZhuYin
 * @since 2024年12月31日
 */
@Slf4j
@Service
@ConditionalOnBean(RocketMqConfig.class)
public class QueueOrderlyConsumer implements MessageListenerOrderly {

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messages, ConsumeOrderlyContext context) {
        try {
            // 获取当前消费的 MessageQueue
            String messageQueue = context.getMessageQueue().toString();
            log.info("Processing messages from MessageQueue: {}", messageQueue);

            for (MessageExt message : messages) {
                // 获取消息所属的主题
                String topic = message.getTopic();
                String messageBody = new String(message.getBody());

                log.info("Received message from topic: {}, message: {}", topic, messageBody);

                // 执行业务逻辑
                processBusinessLogic(topic, messageBody);
            }

            // 如果业务逻辑成功，则返回 SUCCESS
            return ConsumeOrderlyStatus.SUCCESS;
        } catch (Exception e) {
            // 如果业务逻辑失败，返回 SUSPEND_CURRENT_QUEUE_A_MOMENT 表示暂停当前队列一段时间后重试
            log.error("Failed to process messages: {}", e.getMessage());
            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
    }

    private void processBusinessLogic(String topic, String messageBody) throws Exception {
        // 你的业务逻辑处理代码
        // if (result) {
        //     // 模拟成功处理
        //     log.info("Successfully processed message from topic: " + topic);
        // } else {
        //     // 模拟失败处理
        //     throw new RuntimeException("Business logic failed for topic: " + topic);
        // }
    }
}
